package com.atguigu.userprofile.app;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.dao.TagInfoDAO;
import com.atguigu.userprofile.util.MyClickhouseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class TaskBitMap {
    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("task_bitmap");//.setMaster("local[*]");
        SparkContext sparkContext=new SparkContext(conf);

        //获取业务日期
        String busiDate = args[1];

        //1.获取已启用的标签数据
        List<TagInfo> tagInfoListWithOn = TagInfoDAO.getTagInfoListWithOn();

        //2.按照不同标签类型分类
        ArrayList<TagInfo> longList = new ArrayList<>();
        ArrayList<TagInfo> decimalList = new ArrayList<>();
        ArrayList<TagInfo> stringList = new ArrayList<>();
        ArrayList<TagInfo> dateList = new ArrayList<>();

        //3.根据标签类型判断，将相同类型的标签放到一个集合中
        for (TagInfo tagInfo : tagInfoListWithOn) {
            if (ConstCodes.TAG_VALUE_TYPE_LONG.equals(tagInfo.getTagValueType())) {
                longList.add(tagInfo);
            } else if (ConstCodes.TAG_VALUE_TYPE_DECIMAL.equals(tagInfo.getTagValueType())) {
                decimalList.add(tagInfo);
            } else if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagInfo.getTagValueType())) {
                stringList.add(tagInfo);
            } else if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagInfo.getTagValueType())) {
                dateList.add(tagInfo);
            }
        }


        //4.根据不同标签类型，往不同的表里面插入数据
        inserSQLFun(longList, "user_tag_value_long", busiDate);
        inserSQLFun(decimalList, "user_tag_value_decimal", busiDate);
        inserSQLFun(stringList, "user_tag_value_string", busiDate);
        inserSQLFun(dateList, "user_tag_value_date", busiDate);


    }

    private static void inserSQLFun(ArrayList<TagInfo> tagList, String tableName, String busiDate) {
        /**
         * insert into user_tag_value_string
         *   select
         *      tg.1,
         *      if(tg.2='','0',tg.2),
         *      groupBitmapState(cast(uid as Uint64)) uids,
         *      '2020-06-14'
         *   from (
         *       select
         *          arrayJoin([('tg_person_base_gender',tg_person_base_gender),('tg_person_base_agegroup',tg_person_base_agegroup)]) as tg,
         *          uid
         *       from up_tag_merge_20200614) as utg
         *   group by tg.1,tg.2
         */
        if (tagList.size() > 0) {

            //动态拼接Tuple语句
            List<String> tupleList = tagList.stream().map(tagInfo -> "('" + tagInfo.getTagCode().toLowerCase() + "'," + tagInfo.getTagCode().toLowerCase() + ")").collect(Collectors.toList());

            String tupleSQL = StringUtils.join(tupleList, ",");

            //拼接宽表表名
            String mergeTableName = "up_tag_merge_" + busiDate.replace("-", "");

            //保证幂等操作
            String deleteDataSQL = "alter table " + tableName + " delete where dt ='" + busiDate + "'";

            MyClickhouseUtil.executeSql(deleteDataSQL);

            String insertSQL = "insert into " + tableName + "\n" +
                    "            select\n" +
                    "              tg.1,\n" +
                    "              if(tg.2='','0',tg.2),\n" +
                    "              groupBitmapState(cast(uid as UInt64)) uids,\n" +
                    "              '" + busiDate + "'\n" +
                    "            from (\n" +
                    "               select\n" +
                    "                 arrayJoin([" + tupleSQL + "]) as tg,\n" +
                    "                 uid\n" +
                    "             from " + mergeTableName + ") as utg\n" +
                    "          group by tg.1,tg.2";


//            System.out.println(insertSQL);

        MyClickhouseUtil.executeSql(insertSQL);

        }
    }
}
